package com.example.hbase_server;

import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class Utils {
    public static void createTable(Admin admin,String myTableName, String[] colFamily) throws IOException {
        TableName tableName = TableName.valueOf(myTableName);
        if(admin.tableExists(tableName)){
            System.out.println("table is exists!");
        }else {
            TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName);
            for(String str:colFamily){
                ColumnFamilyDescriptor family =
                        ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(str)).build();
                tableDescriptor.setColumnFamily(family);
            }
            admin.createTable(tableDescriptor.build());
        }
    }

    public static boolean judgmentTable(Admin admin,String myTableName) throws IOException {
        TableName tableName = TableName.valueOf(myTableName);
        return admin.tableExists(tableName);
    }

    public static TableResult createCSVTable(StreamTableEnvironment tableEnv,String path,String sql){
        return tableEnv.executeSql(String.format("%s with (" +
                "'connector' = 'filesystem',"+
                "'path' = '%s',"+
                "'csv.ignore-parse-errors' = 'true',"+
                "'format' = 'csv')",sql,path
        ));
    }

    public static TableResult createHbaseTable(StreamTableEnvironment tableEnv,String tableName,String sql){
        return tableEnv.executeSql(String.format("%s with (" +
                "'connector' = 'hbase-2.2',"+
                "'table-name' = '%s',"+
                "'zookeeper.quorum' = 'bigdata:2181')",sql, tableName
        ));
    }

    public static void createAllTable(StreamTableEnvironment tableEnv){
        createHbaseTable(tableEnv,"movies", """
                    create table moviesTable (
                        movieId string primary key not enforced,
                        movieInfo row<title string, genres string>
                    )
            """);
        createHbaseTable(tableEnv,"ratings", """
                    create table ratingsTable (
                        userId_movieId string primary key not enforced,
                        ratingInfo row<rating DOUBLE, ttl BIGINT>
                    )
            """);
        createHbaseTable(tableEnv,"movieRating", """
                    create table movieRatingTable (
                        movieId string primary key not enforced,
                        ratingInfo row<ratingCount DOUBLE, ratingSum DOUBLE,rating DOUBLE>
                    )
            """);
        createHbaseTable(tableEnv,"tempRating", """
                    create table tempRatingTable (
                        ttl BIGINT primary key not enforced,
                        ratingInfo row<rating DOUBLE, movieId string>
                    )
            """);
    }

}
